package com.ehotting.eaf.netty.client;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.ehotting.eaf.core.utils.AssertUtils;
import com.ehotting.eaf.netty.domain.DataAble;
import com.ehotting.eaf.netty.exception.StateConstants;
import com.ehotting.eaf.netty.exception.TcpException;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.Assert;

import java.util.concurrent.TimeUnit;

public abstract class TcpClient implements ChannelFutureListener {
	
	protected Logger logger = LoggerFactory.getLogger(this.getClass());

	protected ChannelInitializer<SocketChannel> channelInitializer;
	
	protected Bootstrap bootstrap;
	
	protected Channel channel;

	protected ChannelFuture future;
	
	protected String ip;
	
	protected Integer port;
	
	public void serverStart() throws Exception {
			try {
				initServer();
				logger.info("ip:{},port:{}",ip,port);
				doConnect();
			} catch (Exception e) {
				e.printStackTrace();
			}
	}

	public void serverStartSync() throws Exception {
		try {
			initServer();
			logger.info("ip:{},port:{}",ip,port);
			this.doConnectSync();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	
	
	public void shutdown(){
		logger.info("Shutdown.");
		channel.close().addListener(new ChannelFutureListener() {
			@Override
			public void operationComplete(ChannelFuture future) throws Exception {
				logger.info("Start operationComplete.");
				future.channel().close();
				if(bootstrap.config()!=null && bootstrap.config().group()!=null){
					bootstrap.config().group().shutdownGracefully();
				}
			}
		});
	}


	public void shutdownFuture(){
		logger.info("Shutdown.");
		channel.closeFuture().addListener(new ChannelFutureListener() {
			@Override
			public void operationComplete(ChannelFuture future) throws Exception {
				logger.info("Start operationComplete.");
				future.channel().close();
				if(bootstrap.config()!=null && bootstrap.config().group()!=null){
					bootstrap.config().group().shutdownGracefully();
				}
			}
		});
	}

	public void initServer() throws Exception {
		bootstrap.group(new NioEventLoopGroup());
		bootstrap.channel(NioSocketChannel.class);
	}
	
	public void setChannelInitializer(ChannelInitializer<SocketChannel> channelInitializer) {
		this.channelInitializer = channelInitializer;
	}

	public void setBootstrap(Bootstrap bootstrap) {
		this.bootstrap = bootstrap;
	}
	
    public void setIp(String ip) {
		this.ip = ip;
	}
	public void setPort(Integer port) {
		this.port = port;
	}

	public void doConnect() throws InterruptedException{
		future = bootstrap.handler(channelInitializer).connect(ip,port).sync();
		future.addListener(this);
	}

	public void doConnectSync() throws InterruptedException{
		future = bootstrap.handler(channelInitializer).connect(ip,port);
		future.addListener(this);
		channel = future.sync().channel();

	}
	
	public void send(String msg) {
		checkChannel();
		channel.writeAndFlush(msg);
   }

	public void send(byte[] msg) {
		checkChannel();
		channel.writeAndFlush(msg);
	}
	
	public void send(DataAble data){
		checkChannel();
		String d = JSON.toJSONString(data, SerializerFeature.DisableCircularReferenceDetect);
		channel.writeAndFlush(d);
	}

	public void checkChannel(){
		AssertUtils.notNull(channel,new TcpException(StateConstants.CHANNEL_NULL,"server[ip:"+ip+",port:"+port+"] not exist,channel is null."));
		AssertUtils.isTrue(channel.isActive(),new TcpException(StateConstants.CHANNEL_INACTIVE,"server[ip:"+ip+",port:"+port+"] channel is inActive."));
	}

	public Channel getChannel() {
		return channel;
	}

	@Override
	public void operationComplete(ChannelFuture future)  {
		if (future.isSuccess()) {
			this.channel = future.channel();
			logger.info("{} Reconnection sucess.",ip+":"+port);
			success();
		} else {
			logger.info("{} Reconnection fail.",ip+":"+port);
			fail();
			//  3秒后重新连接
			future.channel().eventLoop().schedule(() -> {
				try {
					doConnect();
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}, 10, TimeUnit.SECONDS);
		}
	}

	/**
	 * 连接成功的回调
	 */
	public abstract void success();

	/**
	 * 断连时的回调
	 */
	public abstract void fail();

}